/* 
 * Copyright (c) 2009, Giora Kosoi giorak@gmail.com
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *     * Redistributions of source code must retain the above copyright
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright
 *       notice, this list of conditions and the following disclaimer in the
 *       documentation and/or other materials provided with the distribution.
 *     * Neither the name of the project nor the
 *       names of its contributors may be used to endorse or promote products
 *       derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY Giora Kosoi ''AS IS'' AND ANY
 * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL Giora Kosoi BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

#include <sparrow/tcp_io.h>
#include <sparrow/reactor.h>
#include <sparrow/variant_headers.h>
#include <sparrow/util.h>
#include <sparrow/types.h>
#include <sparrow/errors.h>

TcpIo::TcpIo(Reactor *r, const std::string &config)
    :   Handler(r)
{

    WordVector words;
    parse_words(config, words);
    if(words.size() != 4)
    {
        throw_error<std::logic_error>(__FILE__, __LINE__,
                "incorrect number of arguments for TcpIo: config = %s", 
                config.c_str());
    }

    VariantInt32 max_connections(words.at(0));
    max_connections_ = max_connections.uint32();

    VariantInt32 idle_timeout(words.at(1));
    idle_timeout_ = totimespec(idle_timeout.uint32());

    // TODO configurable socket receive buffer 
    VariantInt32 recv_buffer(words.at(2));
    recv_buffer_size_ = recv_buffer.uint32();

    VariantInt32 fail_send_timeout(words.at(3));
    fail_send_timeout_ = fail_send_timeout.uint32();

    register_once(this, &TcpIo::once);
    register_input("new-connection", this, &TcpIo::new_connection);
    register_input("stream-out", this, &TcpIo::stream_out);
    register_input("control-out", this, &TcpIo::control_out);
    register_input("error-out", this, &TcpIo::error_out);
}

TcpIo::~TcpIo()
{
}

void TcpIo::handle_input(int fd, uint64_t token)
{
    get_reactor()->cancel_timer(token);
    get_reactor()->schedule_timer(this, idle_timeout_, token);
    VariantSet *connection_attributes = NULL;
    try{
        connection_attributes = connections_.get(IntKey(token))->variant_set();
        int server_fd = connection_attributes->get("server-fd")->int32();
        if(server_fd != fd)
        {
            throw_error<std::logic_error>(__FILE__, __LINE__,
                    "stored file descriptor for token [%lld] is different from notification",
                    token);
        }
        Octets buffer(recv_buffer_size_, 0);
        ssize_t result = recv(server_fd, &buffer.at(0), buffer.size(), 0);
        if(result == -1)
        {
            if(errno != EINTR && errno != EAGAIN && errno != EWOULDBLOCK)
            {
                throw_error<std::runtime_error>(__FILE__, __LINE__,
                        "failed recv: error: [%s]",
                        strerror(errno));
            }
        }
    else if(result == 0)
    {
        VariantSet *connection_attributes = connections_.release(IntKey(token))->variant_set();
        connection_attributes->set("status", new VariantString("closed"));
        get_reactor()->unregister(token);
        push("control-in", connection_attributes);
    }
        else
        {
            buffer.resize(result);
            VariantSet *v = new VariantSet;
            VariantString *vbuff = new VariantString;
            vbuff->swap(buffer);
            v->set("tcp-chunk-in", vbuff);
            v->set("connection-attributes", connection_attributes->clone());
            push("stream-in", v);
        }
    }
    catch(std::exception &e)
    {
        get_reactor()->unregister(token);
        if(connection_attributes == NULL)
        {
            connection_attributes = new VariantSet();
        }
        connection_attributes->set("error-message", new VariantString(e.what()));
        push("error-in", connection_attributes);
    }
}

void TcpIo::handle_output(int fd, uint64_t token)
{
    get_reactor()->cancel_timer(token);
    get_reactor()->schedule_timer(this, idle_timeout_, token);
    VariantSet *connection_attributes = NULL;
    try
    {
        uint64_t connection_id = token;
        connection_attributes = connections_.get(IntKey(connection_id))->variant_set();
        int32_t server_fd = connection_attributes->get("server-fd")->int32();
        if(server_fd != fd)
        {
            throw_error<std::logic_error>(__FILE__, __LINE__,
                    "stored file descriptor for token [%lld] is different from notification",
                    connection_id);
        }

        SendQueues::iterator qi = send_queues_.find(connection_id);
        if(qi != send_queues_.end())
        {
            Chunk &chunk = qi->second.front();
            int result = send(fd, &chunk.buffer.at(0), chunk.size(), MSG_NOSIGNAL);
            if(result == -1)
            {
                if(errno != EINTR && errno != EAGAIN && errno != EWOULDBLOCK)
                {
                    handle_send_error(connection_id, new VariantSet());
                }
            }
            else if(result < chunk.size())
            {
                chunk.offset += result;
            }
            else if(result == chunk.size())
            {
                qi->second.pop();
                if(qi->second.empty())
                {
                    send_queues_.erase(qi);
                    get_reactor()->unregister_output_handler(connection_id);
                }
            }
        }
        else
        {
            get_reactor()->unregister_output_handler(connection_id);
        }
    }
    catch(std::exception &e)
    {
        get_reactor()->unregister(token);
        if(connection_attributes == NULL)
        {
            connection_attributes = new VariantSet();
        }
        connection_attributes->set("error-message", new VariantString(e.what()));
        push("error-in", connection_attributes);
    }
}

void TcpIo::handle_error(int fd, uint64_t token)
{
}

void TcpIo::handle_timeout(uint64_t token)
{
    printf("timeout %lld\n", token);
    VariantSet *connection_attributes = NULL;
    try
    {
        uint64_t connection_id = token;
        SendQueues::iterator qi = send_queues_.find(connection_id);
        if(qi == send_queues_.end())
        {
            VariantSet *connection_attributes = connections_.release(IntKey(token))->variant_set();
            connection_attributes->set("status", new VariantString("timed-out"));
            get_reactor()->unregister(token);
            push("control-in", connection_attributes);
        }
    }
    catch(std::exception &e)
    {
        get_reactor()->unregister(token);
        if(connection_attributes == NULL)
        {
            connection_attributes = new VariantSet();
        }
        connection_attributes->set("error-message", new VariantString(e.what()));
        push("error-in", connection_attributes);
    }
}

void TcpIo::once(VariantSet *v)
{
}

void TcpIo::new_connection(VariantSet *v)
{
    uint64_t connection_id = get_reactor()->create_token();
    v->set("connection-id", new VariantInt64(connection_id));
    // take ownership of the variant set
    connections_.set(IntKey(connection_id), v);
    int fd = v->get("server-fd")->int32(); 
    get_reactor()->register_input_handler(fd, this, connection_id);
    get_reactor()->schedule_timer(this, idle_timeout_, connection_id);
}

void TcpIo::handle_send_error(uint64_t connection_id, VariantSet *v)
{
    // this connection is in error
    connections_.remove(IntKey(connection_id));
    get_reactor()->unregister(connection_id);
    std::stringstream s;
    s << "failed send: error: [" << strerror(errno) << "]";
    v->set("error-message", new VariantString(s.str()));
    push("error-in", v);
}

void TcpIo::stream_out(VariantSet *v)
{
    uint64_t connection_id = v->get("connection-attributes")->variant_set()->get("connection-id")->int64();
    get_reactor()->cancel_timer(connection_id);
    get_reactor()->schedule_timer(this, idle_timeout_, connection_id);
    Octets chunk;
    v->get("tcp-chunk-out")->swap(chunk);
    SendQueues::iterator qi = send_queues_.find(connection_id);
    if(qi != send_queues_.end())
    {
        // if send queue exists
        // this connection is already registered
        // for output notification
        qi->second.push(Chunk());
        qi->second.back().buffer.swap(chunk);
        qi->second.back().offset = 0;
    }
    else
    {
        VariantSet *connection_attributes = connections_.get(IntKey(connection_id))->variant_set();
        int32_t fd = connection_attributes->get("server-fd")->int32();
        int result = send(fd, &chunk.at(0), chunk.size(), MSG_NOSIGNAL);
        if(result == -1)
        {
            if(errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)
            {
                // send queue for this connection still 
                // does not exist
                std::pair<SendQueues::iterator, bool> 
                    insert_result = send_queues_.insert(std::make_pair(connection_id, SendQueue()));
                // insert_result.second != false b because qi == send_queues_.end()

                SendQueue &q = insert_result.first->second;
                q.push(Chunk());
                q.back().buffer.swap(chunk);
                q.back().offset = 0;
                get_reactor()->register_output_handler(fd, this, connection_id);
            }
            else
            {
                handle_send_error(connection_id, v->clone()->variant_set());
            }
        }
        else if(result < chunk.size())
        {
            std::pair<SendQueues::iterator, bool> 
                insert_result = send_queues_.insert(std::make_pair(connection_id, SendQueue()));
            // insert_result.second != false b because qi == send_queues_.end()
            SendQueue &q = insert_result.first->second;
            q.back().buffer.swap(chunk);
            q.back().offset = result;
            get_reactor()->register_output_handler(fd, this, connection_id);
        }
//        else if(result == chunk.size())
//        {
//            // do nothing
//        }
    }
    delete v;
}

void TcpIo::control_out(VariantSet *v)
{
    const std::string &command = v->get("command")->str();
    if(command.compare("close"))
    {
        close_connection(v);
    }
    else if(command.compare("reset"))
    {
        reset_connection(v);
    }
    else
    {
        uint64_t connection_id = v->get("connection-attributes")->variant_set()->get("connection-id")->int64();
        throw_error<std::logic_error>(__FILE__, __LINE__,
                "not supported command [%s]",
                command.c_str());
    }
}

void TcpIo::error_out(VariantSet *v)
{
    // if there is a need to send an error message
    // use stream-out followed by control-out(command = closed)
    reset_connection(v);
}

void TcpIo::close_connection(VariantSet *v)
{
    uint64_t connection_id = v->get("connection-attributes")->variant_set()->get("connection-id")->int64();
    VariantSet *connection_attributes = connections_.release(IntKey(connection_id))->variant_set();
    int32_t fd = connection_attributes->get("server-fd")->int32();
    get_reactor()->unregister(connection_id);
    close(fd);
    delete connection_attributes;
    delete v;
}

void TcpIo::reset_connection(VariantSet *v)
{
    uint64_t connection_id = v->get("connection-attributes")->variant_set()->get("connection-id")->int64();
    VariantSet *connection_attributes = connections_.release(IntKey(connection_id))->variant_set();
    int32_t fd = connection_attributes->get("server-fd")->int32();
    get_reactor()->unregister(connection_id);
    struct linger l;
    l.l_onoff = 0;
    l.l_linger = 0;
    setsockopt(fd, SOL_SOCKET, SO_LINGER,(void *)&l, sizeof(struct linger));
    close(fd);
    delete connection_attributes;
    delete v;
}
